package com.cn.ycy.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.websocket.server.PathParam;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;


/**
 * sse
 * 服务器单向往客户端推送数据
 */
@Slf4j
@RestController
@RequestMapping("sse")
public class SSEController {

    @GetMapping("/stream")
    public SseEmitter stream(@RequestParam(value = "userId") String userId) {

        log.info("stream start -> {}", 1);

        //建立一个 SSE 实例
        SseEmitter emitter = subscribe(userId);

//        SseEmitter emitter = new SseEmitter();
//        // 在此处通过 emitter 发送实时数据

        log.info("stream end -> {}", 2);

//        emitter.complete();

        return emitter;
    }

    @GetMapping("/sendmsg")
    public String sendmsg(@RequestParam(value = "userId") String userId, @RequestParam("message") String message) {

        log.info("sendmsg start -> {}", 1);

        if (StringUtils.isNotBlank(userId)) {
            SseEmitter push1 = push(userId, message);
            if ("q".equals(message)) {
                // 发送完消息后移除SSE
                removeEmitter(userId);
            }
        } else {
            pushAllUser(message);
        }
        log.info("sendmsg end -> {}", 2);
        return "finish";
    }


    //保存客户连接
    public static Map<String, SseEmitter> userSseEmitters = new HashMap<>();


    /**
     * 用户上线，开启一个SSE连接
     *
     * @param userId
     * @return
     */
    public SseEmitter subscribe(String userId) {
        //创建一个超时时间为30秒的SseEmitter对象
        SseEmitter sseEmitter = new SseEmitter(30 * 1000L);
        //设置回调函数
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        //缓存起来
        userSseEmitters.put(userId, sseEmitter);
        return sseEmitter;
    }

    /**
     * 推送消息给某个客户端
     *
     * @param userId  用户ID
     * @param content 消息体
     */
    public static SseEmitter push(String userId, String content) {
        SseEmitter sseEmitter = userSseEmitters.get(userId);
        if (sseEmitter != null) {
            try {
                sseEmitter.send(content);
            } catch (Exception e) {
                log.error("用户-{} 推送消息异常：{}", userId, e);
            }
        } else {
            log.info("用户-{} 连接不存在", userId);
        }
        return sseEmitter;
    }

    /**
     * 推送消息给所有客户端
     *
     * @param content 消息体
     */
    public static void pushAllUser(String content) {
        for (SseEmitter emitter : userSseEmitters.values()) {
            try {
                emitter.send(SseEmitter.event().name("全局消息").data(content));
            } catch (Exception e) {
                log.error("推送消息异常：{}", e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * 删除某个客户连接
     *
     * @param userId
     */
    public static void removeEmitter(String userId) {
        SseEmitter sseEmitter = userSseEmitters.get(userId);
        if (sseEmitter != null) {
            sseEmitter.complete();
            userSseEmitters.remove(userId);
        } else {
            log.info("用户-{} 连接不存在", userId);
        }
    }


    private Runnable completionCallBack(String userId) {
        return () -> {
            log.info("用户-{} 发送完毕", userId);
        };
    }

    /**
     * 出现超时，将当前用户缓存删除
     *
     * @param userId
     * @return
     */
    private Runnable timeoutCallBack(String userId) {
        return () -> {
            log.info("用户-{} 连接超时", userId);
            userSseEmitters.remove(userId);
        };
    }

    /**
     * 出现异常，将当前用户缓存删除
     *
     * @param userId
     * @return
     */
    private Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            log.info("用户-{} 连接异常", userId);
            userSseEmitters.remove(userId);
        };
    }

}
